package com.vip.websocket.service;

import com.vip.websocket.util.SpringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;

/**
 * WebSocket的具体实现类
 *
 * @author wgb
 * @date 2020/4/11 17:14
 */
@Slf4j
@Component
@ServerEndpoint(value = "/{topic}/{id}")
public class WebSocketServer {
    /**
     * 客户端ID
     */
    private String id = "";

    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;

    /**
     * 记录当前在线连接数(为保证线程安全，须对使用此变量的方法加lock或synchronized)
     */
    private static int onlineCount = 0;

    /**
     * 用来存储当前在线的客户端(此map线程安全)
     */
    private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    /**
     * 因为@ServerEndpoint不支持注入，所以使用SpringUtils获取IOC实例
     */
    private StringRedisTemplate redisTemplate = SpringUtils.getBean(StringRedisTemplate.class);
    /**
     * Redis监听器容器
     */
    private RedisMessageListenerContainer redisMessageListenerContainer = SpringUtils.getBean(RedisMessageListenerContainer.class);

    /**
     * 连接建立成功后调用
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("topic") String topic, @PathParam(value = "id") String id) {
        this.session = session;
        // 接收到发送消息的客户端ID
        this.id = id;
        // 加入map中
        webSocketMap.put(id, this);
        // 在线数加1
        addOnlineCount();
        log.info("客户端" + id + "加入，当前在线数为：" + getOnlineCount());
        SubscribeService subscribeService = new SubscribeService();
        subscribeService.setSession(session);
        subscribeService.setStringRedisTemplate(redisTemplate);
        //客户端信息存放到Redis中
        redisTemplate.opsForValue().set(String.format("%s:%s", topic, id), id);
        //设置订阅topic
        redisMessageListenerContainer.addMessageListener(subscribeService, new ChannelTopic(String.format("%s:%s", topic, id)));
        try {
            // TODO 连接成功后操作
            sendMessage("WebSocket连接成功");
        } catch (IOException e) {
            log.error("WebSocket IO异常");
        }
    }

    /**
     * 连接关闭时调用
     */
    @OnClose
    public void onClose(Session session, @PathParam("topic") String topic, @PathParam("id") String id) {
        log.info(session.getId() + "java websocket:关闭连接");
        redisTemplate.delete(String.format("%s:%s", topic, id));
        // 从map中删除
        webSocketMap.remove(this);
        // 在线数减1
        subOnlineCount();
        log.info("有一连接关闭，当前在线数为：" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用
     *
     * @param message 客户端发送过来的消息：消息格式：内容 - 表示群发，内容|X - 表示发给id为X的客户端
     * @param session
     */
    @OnMessage
    public void onMessage(Session session, String message, @PathParam("topic") String topic, @PathParam("id") String id) {
        PublishService publishService = SpringUtils.getBean(PublishService.class);
        publishService.publish(topic, message);
        log.info("来自客户端的消息:" + message);
        String[] messages = message.split("[|]");
        try {
            if (messages.length > 1) {
                sendToUser(messages[0], messages[1]);
            } else {
                sendToAll(messages[0]);
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 发生错误时回调
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("WebSocket发生错误");
        error.printStackTrace();
    }

    /**
     * 推送信息给指定ID客户端，如客户端不在线，则返回不在线信息给自己
     *
     * @param message      客户端发来的消息
     * @param sendClientId 客户端ID
     * @throws IOException
     */
    public void sendToUser(String message, String sendClientId) throws IOException {
        if (webSocketMap.get(sendClientId) != null) {
            if (!id.equals(sendClientId)) {
                webSocketMap.get(sendClientId).sendMessage("客户端" + id + "发来消息：" + " <br/> " + message);
            } else {
                webSocketMap.get(sendClientId).sendMessage(message);
            }
        } else {
            // 如客户端不在线，则返回不在线信息给自己
            sendToUser("当前客户端不在线", id);
        }
    }

    /**
     * 推送发送信息给所有人
     *
     * @param message 要推送的消息
     * @throws IOException
     */
    public void sendToAll(String message) throws IOException {
        for (String key : webSocketMap.keySet()) {
            webSocketMap.get(key).sendMessage(message);
        }
    }

    /**
     * 推送消息
     *
     * @param message 要推送的消息
     * @throws IOException
     */
    private void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    private static synchronized int getOnlineCount() {
        return onlineCount;
    }

    private static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    private static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }

    public Session getSession() {
        return session;
    }

    public void setSession(Session session) {
        this.session = session;
    }
}
